package com.gitee.mocoffee.flow;

import java.util.ArrayList;
import java.util.List;
import java.util.Map;
import java.util.concurrent.Callable;
import java.util.concurrent.ConcurrentHashMap;
import java.util.concurrent.ExecutorService;
import java.util.concurrent.Executors;
import java.util.concurrent.Future;
import java.util.concurrent.TimeUnit;
import java.util.stream.Collectors;
import java.util.stream.Stream;

import javax.annotation.PostConstruct;
import javax.annotation.PreDestroy;

import lombok.AccessLevel;
import lombok.AllArgsConstructor;
import lombok.EqualsAndHashCode;
import lombok.Getter;
import lombok.NonNull;
import lombok.RequiredArgsConstructor;
import lombok.Setter;
import lombok.ToString;
import lombok.experimental.FieldDefaults;
import lombok.extern.slf4j.Slf4j;

/**
 * 一个简单的业务流程器.它包含引用了一系列的业务处理器,按照一定的顺序进行串行或者并行处理.
 * 对于有并行处理的场景,需要外部提供一个线程池,或者提供线程数后自动构建一个线程池.
 * 注:Flow对象需要有生命周期的管理,参考{@link #init()}和{@link #destroy()}方法
 *
 * @author lijia
 */
@AllArgsConstructor
@RequiredArgsConstructor
@FieldDefaults(level = AccessLevel.PRIVATE)
@Slf4j
@Getter
public class Flow<INPUT> {
    /**
     * 处理器集合,是一个二维结构.第一维串行处理,第二维并行处理
     */
    @NonNull
    List<Processor<INPUT>[]> chain;

    /**
     * 业务异常处理器,如不提供,会使用默认的{@link com.gitee.mocoffee.flow.ProcessExceptionHandler.DefaultProcessExceptionHandler}
     */
    @Setter
    ProcessExceptionHandler<INPUT> exceptionHandler;

    /**
     * 外部提供的线程池,如不提供,那必须定义{@link #threadNum},以便自动创建一个内部线程池
     */
    @Setter
    ExecutorService threadPool;

    /**
     * 在不提供外部线程池的情况下,使用此线程数构建一个内部的线程池
     */
    @Setter
    int threadNum;

    volatile boolean innerPool = false;//是否内部线程池
    volatile boolean inited    = false;
    volatile boolean shutdown  = false;

    /**
     * 并行处理时,将业务处理器适配为Callable对象
     *
     * @param <INPUT>
     */
    @AllArgsConstructor
    @FieldDefaults(level = AccessLevel.PRIVATE)
    private static class ProcessorCallable<INPUT> implements Callable<Boolean> {
        Processor<INPUT>                  processor;
        INPUT                             input;
        ConcurrentHashMap<String, Object> context;

        @Override
        public Boolean call() throws Exception {
            return processor.process(input, context);
        }
    }

    /**
     * 封装类,业务流程器的处理结果
     */
    @AllArgsConstructor
    @FieldDefaults(level = AccessLevel.PRIVATE)
    @Getter
    @ToString
    @EqualsAndHashCode
    public static class FlowResult {
        boolean                           success;
        ConcurrentHashMap<String, Object> context;
    }

    /**
     * 业务流程器初始化方法
     */
    @SuppressWarnings("unchecked")
	@PostConstruct
    public void init() {
        if (log.isDebugEnabled())
            log.debug("init()");

        if (inited)
            return;

        if (chain == null || chain.size() == 0)
            throw new IllegalArgumentException("处理链[chain]必须非空");

        boolean needConcurrent = false;//是否需要并行处理
        for (Processor<INPUT>[] ps : chain) {
            if (ps == null || ps.length == 0)
                throw new IllegalArgumentException("处理链[chain]中不能含有空处理器");
            if (ps.length > 1)
                needConcurrent = true;
        }

        if (needConcurrent && threadPool == null) {
            if (threadNum < 1)
                throw new IllegalArgumentException("未配置外部线程池[threadPool],线程数[threadNum]设置必须大于0");
            threadPool = Executors.newFixedThreadPool(threadNum);
            innerPool = true;
            log.info("已配置内部线程池,线程数{}", threadNum);
        }

        if (exceptionHandler == null)
            exceptionHandler = ProcessExceptionHandler.DefaultProcessExceptionHandler.me;

        inited = true;
    }

    /**
     * 业务流程器的销毁方法
     *
     * @throws InterruptedException
     */
    @PreDestroy
    public void destroy() {
        if (log.isDebugEnabled())
            log.debug("init()");

        if (shutdown)
            return;

        if (threadPool != null && innerPool) {
            log.info("正在优雅关闭内部线程池,超时设置为10秒钟");
            threadPool.shutdown();
            try {
                if (threadPool.awaitTermination(10, TimeUnit.SECONDS))
                    log.info("内部线程池已优雅关闭");
                else
                    log.warn("内部线程池未能优雅关闭");
            } catch (InterruptedException e) {
                threadPool.shutdownNow();
                Thread.currentThread().interrupt();
            }
        }
        shutdown = true;
    }

    /**
     * 业务流程器启动入口
     *
     * @param input
     * @return
     */
    public FlowResult run(INPUT input) {
        Processor<INPUT>[] ps = null;
        boolean success = true;
        Map<Future<Boolean>, Processor<INPUT>> futureMap = null;
        ConcurrentHashMap<String, Object> context = new ConcurrentHashMap<>(16, 1.0f);
        int faultIndex = -1;

        int chainSize = chain.size();
        for (int i = 0; i < chainSize; i++) {
            ps = chain.get(i);
            if (ps.length == 1)//对于单个处理器按照一般的串行处理
                try {
                    success = ps[0].process(input, context);
                    if (!success) {
                        faultIndex = i;
                        break;
                    }
                } catch (Exception e) {
                    exceptionHandler.onProcessException(ps[0], input, context, e);
                    success = false;
                    faultIndex = i;
                    break;
                }
            else {//对于并行处理器,全部成功才算成功,否则成功的都要回滚
                futureMap = Stream.of(ps).collect(Collectors.toMap(
                        p -> threadPool.submit(new ProcessorCallable<INPUT>(p, input, context)),
                        p -> p));

                List<Processor<INPUT>> successful = new ArrayList<>(ps.length);

                for (Map.Entry<Future<Boolean>, Processor<INPUT>> entry : futureMap.entrySet()) {
                    try {
                        boolean r = entry.getKey().get();

                        if (r)
                            successful.add(entry.getValue());

                        success = success && r;
                    } catch (Exception e) {
                        exceptionHandler.onProcessException(entry.getValue(), input, context, e);
                        success = false;
                    }
                }

                /*
                如果不是全部成功,那么已经成功的处理器需要回滚
                 */
                if (!success) {
                    rollbackList(successful, input, context);
                    faultIndex = i;
                    break;
                }
            }
        }

        /*
        退出循环后如果发现不成功,说明中途发生了异常,那么从faultIndex向前都需要回滚
         */
        if (!success)
            rollbackChain(faultIndex, input, context);

        return new FlowResult(success, context);
    }

    /**
     * 从faultIndex(不包括)向前的处理器进行回滚操作
     *
     * @param faultIndex
     * @param input
     * @param context
     */
    private void rollbackChain(int faultIndex, INPUT input, ConcurrentHashMap<String, Object> context) {
        for (int i = faultIndex; i > -1; i--)
            rollbackArray(chain.get(i), input, context);
    }

    /**
     * 回滚处理器数组
     *
     * @param processors
     * @param input
     * @param context
     */
    private void rollbackArray(Processor<INPUT>[] processors, INPUT input,
                               ConcurrentHashMap<String, Object> context) {
        for (Processor<INPUT> p : processors)
            try {
                p.rollback(input, context);
            } catch (Exception e) {
                exceptionHandler.onRollbackException(p, input, context, e);
            }
    }

    /**
     * 回滚处理器列表
     *
     * @param processors
     * @param input
     * @param context
     */
    private void rollbackList(List<Processor<INPUT>> processors, INPUT input,
                              ConcurrentHashMap<String, Object> context) {
        processors.forEach(p -> {
            try {
                p.rollback(input, context);
            } catch (Exception e) {
                exceptionHandler.onRollbackException(p, input, context, e);
            }
        });
    }
}
